package net.bwie.realtime.dws.log.function;

import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import net.bwie.realtime.dws.log.bean.MonitorInfo;
import net.bwie.realtime.dws.log.utils.JdbcUtils;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.configuration.Configuration;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.concurrent.TimeUnit;

public class CarSpeedingFilterFunction extends RichFilterFunction<MonitorInfo> {

    Connection connection;
    PreparedStatement ps;
    ResultSet rs;
    LoadingCache<String, Integer> cache;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = JdbcUtils.getconnection();
        ps = connection.prepareStatement("select limit_speed from monitor_limit_speed where monitor_id = ?");
        cache = cache = CacheBuilder.newBuilder()
                .maximumSize(100)                                   //缓存最大存储条数
                .expireAfterWrite(100, TimeUnit.MINUTES)    //缓存过期时间
                .build(new CacheLoader<String, Integer>() {
                    // load 是在 别人问cache 要数据，假如缓存没有，就会触发 load 执行
                    @Override
                    public Integer load(String monitorId) throws Exception {
//                        System.out.println("查询数据库.......");
                        ps.setString(1, monitorId);
                        rs = ps.executeQuery();
                        //如果t_monitor_info无法查询出该卡口的编号，则给定一个60的限速
                        int limit_speed = 60;
                        if (rs.next()) {
                            limit_speed = rs.getInt("limit_speed");
                        }
                        return limit_speed;
                    }
                });
    }

    @Override
    public void close() throws Exception {
        JdbcUtils.release(rs, ps, connection);
    }

    @Override
    public boolean filter(MonitorInfo value) throws Exception {
        Integer limit_speed = cache.get(value.getMonitorId());
        value.setLimitSpeed(limit_speed);
        return value.getSpeed() > limit_speed * 1.1; //超速10%，判定为超速
    }
}
